package de.kardass.jannobatch.manager;

import static de.kardass.jannobatch.manager.ReflectUtil.*;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;

import de.kardass.jannobatch.errorhandler.AbortBatchException;
import de.kardass.jannobatch.errorhandler.ErrorHandler;
import de.kardass.jannobatch.errorhandler.ErrorHandlerReturnCode;
import de.kardass.jannobatch.errorhandler.HandledError;
import de.kardass.jannobatch.errorhandler.SkipItemException;
import de.kardass.jannobatch.ItemProcessor;
import de.kardass.jannobatch.ItemReader;
import de.kardass.jannobatch.ItemWriter;


public class BatchWorkflow {

	protected final ItemReader<?> itemReader;
	
	protected final List<ItemProcessor<?,?>> itemProcessors;
	
	protected final ItemWriter<?> itemWriter;
	
	protected final HashMap<Class<? extends Throwable>, ErrorHandler<?, ? extends Throwable>> errorHandlers;
	
	protected final boolean itemReaderExists;
	protected final boolean itemProcessorExists;
	protected final boolean itemWriterExists;
	
	
	
	/**
	 * @param reader
	 * @param processors
	 * @param writer
	 * @param errorHandlers
	 */
	public BatchWorkflow(
			ItemReader<?> reader,
			List<ItemProcessor<?,?>> processors,
			ItemWriter<?> writer,
			HashMap<Class<? extends Throwable>, ErrorHandler<?, ? extends Throwable>> errorHandlers) {
		super();
		this.itemReader = reader;
		this.itemProcessors = processors;
		this.itemWriter = writer;
		this.itemReaderExists = this.itemReader != null;
		this.itemProcessorExists = this.itemProcessors != null && this.itemProcessors.size() > 0;
		this.itemWriterExists = this.itemWriter != null;
		this.errorHandlers = errorHandlers;
	}
	
	/**
	 * Starts processing this batch workflow. 
	 */
	public void process() {
	    try {
	        startUp();
	        processItems();
        } finally {
            shutDown();
        }
	}
	
	/**
	 * Processes all items: reads each item and sends them through the process 
	 * chain and the final writer.
	 */
	private void processItems() {
		if (itemReaderExists) {
			final Iterator<?> iterator = itemReader.iterator();
			while (iterator.hasNext()) {
				try {
                    runWorkflowFor(iterator.next());
                } catch (SkipItemException e) {
                    // OK, as skip item and process next item
                }
			}
		} else {
            try {
                runWorkflowFor(null);
            } catch (SkipItemException e) {
                // OK, as skip item and process next item
            }
		}
	}
	
	/**
	 * Runs the complete process chain incl. the final writer, if available.
	 * @param <T>
	 * @param item
	 * @throws SkipItemException 
	 */
	private <T> void runWorkflowFor(T item) throws SkipItemException {
		final Object outItem = runSteps(0, item);
		if (itemWriterExists) {
			try {
                invokeWrite(itemWriter, outItem);
            } catch (Throwable e) {
                throw new RuntimeException(e.getCause());
            }
		}
	}
	
	/**
	 * Runs the process chain (processors) from 'position' in chain with the given item.
	 * @param position
	 * @param input
	 * @return
	 * @throws SkipItemException 
	 */
	private Object runSteps(int position, Object input) throws SkipItemException {
		if (position < this.itemProcessors.size()) {
            try {
                final Object output = invokeProcess(itemProcessors.get(position), input);
                ++position;
                return runSteps(position, output);
            } catch (Throwable t) {
                final HandledError<?, ?> handledError = handleError(input, t);
                final ErrorHandlerReturnCode rc = handledError.getReturnCode();
                if (rc.equals(ErrorHandlerReturnCode.REPEAT_CURRENT_STEP)) {
                    return runSteps(position, input);
                } else if (rc.equals(ErrorHandlerReturnCode.CONTINUE_WITH_NEXT_STEP)) {
                    ++position;
                    return runSteps(position, input);
                } else if (rc.equals(ErrorHandlerReturnCode.CONTINUE_WITH_NEXT_ITEM)) {
                    throw new SkipItemException("Skipping item " + input);
                } else {
                    throw new AbortBatchException("Aborting batch due error while processing " + input, handledError.getException());
                }
            }
		} else {
			return input;
		}
	}
	
	/**
	 * Calls the processor's 'process' method with the input as parameter. 
	 * @param processor
	 * @param input
	 * @return
	 * @throws Throwable 
	 */
	private Object invokeProcess(Object processor, Object input) throws Throwable {
		try {
			final Method method = getProcessMethodWithParameter(processor.getClass(), input.getClass());
			return method.invoke(processor, input);
		} catch (InvocationTargetException e) {
		    throw e.getCause();
		}
	}
	
	/**
	 * Calls the processor's 'process' method with the input as parameter. 
	 * @param processor
	 * @param input
	 * @return
	 * @throws Throwable 
	 */
	private Object invokeWrite(Object processor, Object input) throws Throwable {
		try {
			final Method method = getWriteMethodWithParameter(processor.getClass(), input.getClass());
			return method.invoke(processor, input);
        } catch (InvocationTargetException e) {
            throw e.getCause();
        }
	}
	
	/**
	 * Handels an error for the given exception with an erro handler (if available)
	 * for the given item. 
	 * @param <Item>
	 * @param item
	 * @param t
	 */
	private <Item, T extends Throwable> HandledError<Item, T> handleError(Item item, T t) {
		final ErrorHandler<Item, T> errorHandler = findErrorHandler(t.getClass());
		if (errorHandler == null) {
			throw new RuntimeException("No error handler found for exception " + t);
		} else {
			return errorHandler.handleException(item, t);
		}
	}
	
	/**
	 * Tries to find an error handler for the given throwable (incl. it's super classes)
	 * @param t
	 * @return
	 */
    @SuppressWarnings("unchecked")
    private <Item, T extends Throwable> ErrorHandler<Item, T> findErrorHandler(Class<?> t) {
		// try to find a direkt 'linked' hadler
		ErrorHandler<Item, T> beh = (ErrorHandler<Item, T>) errorHandlers.get(t);
		// if not found, try to find a handler from the super classes
		if (beh == null) {
			final Class<?> superClass = t.getSuperclass();
			if (superClass != null) {
				beh = findErrorHandler(superClass);
			}
		}
		return beh;
	}

	/**
	 * Initializes all parts of the process chain (reader, processors and writer)
	 */
	private void startUp() {
		if (itemReaderExists) {
			itemReader.onStartUp();
		}
		if (itemProcessorExists) {
			for (ItemProcessor<?,?> aProc : itemProcessors) {
				aProc.onStartUp();
			}
		}
		if (itemWriterExists) {
			itemWriter.onStartUp();
		}
	}

	/**
	 * Shuts down all parts of the process chain (reader, processors and writer)
	 */
	private void shutDown() {
		if (itemReaderExists) {
			itemReader.onShutDown();
		}
		if (itemProcessorExists) {
			for (ItemProcessor<?,?> aProc : itemProcessors) {
				aProc.onShutDown();
			}
		}
		if (itemWriterExists) {
			itemWriter.onShutDown();
		}
	}

	public boolean itemReaderExists() {
		return itemReaderExists;
	}

	public boolean itemProcessorExists() {
		return itemProcessorExists;
	}

	public boolean itemWriterExists() {
		return itemWriterExists;
	}

	public ItemReader<?> getItemReader() {
		return itemReader;
	}

	public List<ItemProcessor<?, ?>> getItemProcessors() {
		return itemProcessors;
	}

	public ItemWriter<?> getItemWriter() {
		return itemWriter;
	}

}
